package com.iot.mqtt.service.Impl;
import com.aliyun.iot.as.bridge.core.BridgeBootstrap;
import com.aliyun.iot.as.bridge.core.handler.DownlinkChannelHandler;
import com.aliyun.iot.as.bridge.core.handler.UplinkChannelHandler;
import com.aliyun.iot.as.bridge.core.model.ProtocolMessage;
import com.aliyun.iot.as.bridge.core.model.Session;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.iot.mqtt.service.AliCloud;
import com.iot.mqtt.service.MqttService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Service("AliCloudImpl")
public class AliCloudImpl implements AliCloud {
    private static BridgeBootstrap bridgeBootstrap;

    private static UplinkChannelHandler uplinkChannelHandler;

    private final static ExecutorService executorService  = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors() * 2,
            60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000),
            new ThreadFactoryBuilder().setDaemon(true).setNameFormat("bridge-downlink-handle-%d").build(),
            new ThreadPoolExecutor.AbortPolicy());

    public UplinkChannelHandler getUplinkChannelHandler(){
        return uplinkChannelHandler;
    }

    public static void subTopics() {
        bridgeBootstrap = new BridgeBootstrap();
        bridgeBootstrap.bootstrap(new DownlinkChannelHandler() {
            @Override
            public boolean pushToDevice(Session session, String topic, byte[] payload) {
                //从云端获取下行消息
                executorService.submit(() -> handleDownLinkMessage(session, topic, payload));
                return true;
            }

            @Override
            public boolean broadcast(String s, byte[] bytes) {
                return false;
            }
        });
        uplinkChannelHandler = new UplinkChannelHandler();
    }

    public static void sendMsg(String originalIdentity,String topic,String mqttMessage){
        //向主题发送消息
        ProtocolMessage protocolMessage = new ProtocolMessage();
        protocolMessage.setPayload(mqttMessage.getBytes());
        protocolMessage.setQos(0);
        protocolMessage.setTopic("/"+topic);
        uplinkChannelHandler.doPublishAsync(originalIdentity, protocolMessage);
    }

    public static void connected(String originalIdentity){
        //特定协议的设备会话，如果没有，则只能使用空对象
        Object originalChannel = new Object();
        Session session = Session.newInstance(originalIdentity, originalChannel);
        //设备上线
        UplinkChannelHandler uplinkChannelHandler = new UplinkChannelHandler();
        uplinkChannelHandler.doOnline(session, originalIdentity);
    }

    public static void disconnected(String originalIdentity){
        //设备下线
        UplinkChannelHandler uplinkChannelHandler = new UplinkChannelHandler();
        uplinkChannelHandler.doOffline(originalIdentity);
    }

    private static void handleDownLinkMessage(Session session, String topic, byte[] payload) {
        String content = new String(payload);
        System.out.println("接收到阿里云平台发送的信息, session:"+session+", topic:"+topic+", content:"+content);
        MqttServiceImpl.sendToDevice(topic,content);
    }
}
